package com.ysj.demo.rxjavatest.testUntil;

import android.util.Log;

import com.ysj.demo.rxjavatest.bean.Person;
import com.ysj.demo.rxjavatest.bean.Plan;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

/**
 * Created by 80530 on 2018/6/13.
 */

public class Test {
    private static final String TAG = "MainActivity";

    //分开创建
    public static void firstTest() {
        //创建被观察者
        Observable observable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "============================currentThreadname:" + Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        });

        //创建观察者
        Observer observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "============================onSubscribe");
            }

            @Override
            public void onNext(Integer o) {
                Log.d(TAG, "============================onNext" + o);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "============================onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "============================onComplete");
            }
        };

        //订阅
        observable.subscribe(observer);
    }

    //链式创建
    public static void second() {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> e) throws Exception {
                Log.d(TAG, "============================currentThreadname:" + Thread.currentThread().getName());
                e.onNext(1);
                e.onNext(2);
                e.onNext(3);
                e.onComplete();
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "============================onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "============================onNext" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "============================onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "============================onComplete");
            }
        });
    }

    //just操作,可以传入最多十个参数，是否是按顺序打印？
    public static void three() {
        Observable.just(1, 2, 3).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "============================onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "============================onNext" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "============================onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "============================onComplete");
            }
        });
    }

    //from操作，传入数组，可以超过10个变量
    public static void four() {
        Integer[] array = {1, 2, 3, 4};
        Observable.fromArray(array).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.d(TAG, "============================onSubscribe");
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "============================onNext" + integer);
            }

            @Override
            public void onError(Throwable e) {
                Log.d(TAG, "============================onError");
            }

            @Override
            public void onComplete() {
                Log.d(TAG, "============================onComplete");
            }
        });
    }

    //具有返回值
    public static void five() {
        Observable.fromCallable(new Callable<Integer>() {
            @Override
            public Integer call() throws Exception {
                return 1;
            }
        }).subscribe(new Consumer<Integer>() {
            @Override
            public void accept(Integer integer) throws Exception {
                Log.d(TAG, "============================accept 返回值" + integer);
            }
        });
    }


    //fromFuture
    public static void six() {
        final FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Log.d(TAG, "Callable is running");
                return "返回结果";
            }
        });

        Observable.fromFuture(futureTask).doOnSubscribe(new Consumer<Disposable>() {
                                                            @Override
                                                            public void accept(Disposable disposable) throws Exception {
                                                                futureTask.run();
                                                            }
                                                        }
        ).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Exception {
                Log.d(TAG, "============================accept 返回值" + s);
            }
        });
    }

    //fromIterable list集合
    public static void seven() {
        List<Integer> list = new ArrayList<>();
        list.add(0);
        list.add(1);
        list.add(2);
        list.add(3);
        Observable.fromIterable(list)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d(TAG, "=================onSubscribe");
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.d(TAG, "=================onNext " + integer);
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d(TAG, "=================onError ");
                    }

                    @Override
                    public void onComplete() {
                        Log.d(TAG, "=================onComplete ");
                    }
                });
    }

    //defer 这个方法的作用就是直到被观察者被订阅后才会创建被观察者。
    public static void eight() {
        // i 要定义为成员变量
        Integer i = 100;
        Observable<Integer> observable = Observable.defer(new Callable<ObservableSource<? extends Integer>>() {
            @Override
            public ObservableSource<? extends Integer> call() throws Exception {
                return Observable.just(100);
            }
        });
        i = 200;
        Observer observer = new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
            }

            @Override
            public void onNext(Integer integer) {
                Log.d(TAG, "================onNext " + integer);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        };
        observable.subscribe(observer);
        i = 300;
        observable.subscribe(observer);
    }

    //map  类型转换 可以将被观察者发送的数据类型转变成其他的类型
    public static void nine() {
        Observable.just(1, 2, 3).map(new Function<Integer, String>() {
            @Override
            public String apply(Integer integer) throws Exception {
                return "I'm " + integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                Log.e(TAG, "===================onSubscribe");
            }

            @Override
            public void onNext(String s) {
                Log.e(TAG, "===================onNext " + s);
            }

            @Override
            public void onError(Throwable e) {
            }

            @Override
            public void onComplete() {
            }
        });
    }

    //flatMap() 其实与 map() 类似，但是 flatMap() 返回的是一个 Observerable。现在用一个例子来说明 flatMap() 的用法。
    public static void ten(int type) {
        List personList = new ArrayList();
        switch (type) {
            //map实现
            case 1:
                Observable.fromIterable(personList)
                        .map(new Function<Person, List<Plan>>() {
                            @Override
                            public List<Plan> apply(Person person) throws Exception {
                                return person.getPlanList();
                            }
                        })
                        .subscribe(new Observer<List<Plan>>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                            }

                            @Override
                            public void onNext(List<Plan> plans) {
                                for (Plan plan : plans) {
                                    List<String> planActionList = plan.getActionList();
                                    for (String action : planActionList) {
                                        Log.d(TAG, "==================action " + action);
                                    }
                                }
                            }

                            @Override
                            public void onError(Throwable e) {
                            }

                            @Override
                            public void onComplete() {
                            }
                        });
                break;
            //flatMap实现
            case 2:
                Observable.fromIterable(personList)
                        .flatMap(new Function<Person, ObservableSource<Plan>>() {
                            @Override
                            public ObservableSource<Plan> apply(Person person) {
                                return Observable.fromIterable(person.getPlanList());
                            }
                        })
                        .flatMap(new Function<Plan, ObservableSource<String>>() {
                            @Override
                            public ObservableSource<String> apply(Plan plan) throws Exception {
                                return Observable.fromIterable(plan.getActionList());
                            }
                        })
                        .subscribe(new Observer<String>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                            }

                            @Override
                            public void onNext(String s) {
                                Log.d(TAG, "==================action: " + s);
                            }

                            @Override
                            public void onError(Throwable e) {
                            }

                            @Override
                            public void onComplete() {
                            }
                        });
                break;
        }
    }

}
